package parking.abcast;

import java.io.Serializable;
import java.rmi.RemoteException;
import java.rmi.server.UnicastRemoteObject;
import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedList;
import java.util.Set;
import java.util.logging.Level;
import java.util.logging.Logger;

import parking.CarManager;
import parking.Door;
import parking.IAlgorithm;
import parking.LogManager;
import parking.Message;
import reso.Client;
import reso.Reso;

/**
 * Class implementing the ABCast protocol
 * The principle is simple in ABCast and effective. Suppose we have a process who wants to apply for an entry 
 * into critical section. It will concatenates the value of its internal clock with its ID (e.g 3.1) and 
 * increment it's own internal clock. Afterwards, will diffuse this information to all the other processes 
 * within the group including itself. On receipt of this information, a process will increment its own internal
 * clock and again concatenates this number with its ID (e.g 5.3) and forward this new value to the sender of 
 * the message. The received message by the process is marked as "PENDING" and kept in a sorted list in ascending 
 * order of calculated values.
 * The author of the original message will wait till it receives all the values of all the processes within 
 * the group.On reception of all the values, the author will calculate the maximum value amongst all the received
 * values and will again diffuse this final calculated value. On reception of a final calculated value, a process
 * will update the new calculated value of the message marked previously as "PENDING" and will change too the 
 * status to "READY". If the message is found to be on top of the list, as it has the "READY" status it is
 * delivered and it can be treated.
 * @author Sophie and Ronnie
 * @date 9 November 2013
 *
 */
public class ABCast extends UnicastRemoteObject implements IAlgorithm, Client, Runnable {
	private static final long serialVersionUID = -3054098230317709938L;
	private Door myDoor;									//instance of a door
	private Reso myReso;									//distant object for communication between doors
	private LinkedList<MessageDiffusion> listeMessages;		//list of all received messages
	private int estampilleLocale;							//internal clock
	private Set<Integer> identifiantsPortes;				//a set of all the identities of the doors in the group
	private ArrayList<RetourAck> gestionnaireRetourAck;		//dynamic table containing for each message, acknowledgments of all the other processes
	private LogManager myLogManager;						//object containing the logger
	private Logger myLogger;								//logger allowing to write in a file the execution trace of the algorithm
	private CarManager carManager;
	
	/**
	 * Constructor for the ABCast class
	 * @param door : The door executing the algorithm
	 * @param nbSpots : Number of spots within the parking
	 * @param myReso : Distant object for the exchange of messages between doors
	 * @throws RemoteException : Exceptions are thrown if distant objects aren't found
	 */
	public ABCast(Door door, int nbSpots, Reso myReso, CarManager carManager) throws RemoteException {
		super();
		this.myDoor = door;
		//initialization of the local clock
		this.estampilleLocale = 1;
		this.gestionnaireRetourAck = new ArrayList<RetourAck>();
		this.listeMessages = new LinkedList<MessageDiffusion>();
		this.myReso = myReso;
		this.carManager = carManager;
		//attribution of an ID to the door
		this.myDoor.setIdentity(this.myReso.declareClient(this));
		
		//initialization of the log manager with its appropriate level (info for normal outputs, config for debug process)
		this.myLogManager = new LogManager("../logParking/Protocole_ABCast/" + "Log_AB" + this.myDoor.getIdentity() + ".log", Level.INFO);
		this.myLogger = this.myLogManager.getlogger();
		this.myLogger.config("Identité porte : " + this.myDoor.getIdentity());
	}
	
	/**
	 * This method is used to diffuse a message within the group. It can be an entry or an exit for a car.
	 * @param entreeSortie : 0 representing an entry of a car, 1 an exit
	 */
	public synchronized void envoieDiffusionGroupe(int entreeSortie) {
		//creation of the message to diffuse
		MessageDiffusion msgDiffusion = new MessageDiffusion(entreeSortie, estampilleLocale, myDoor.getIdentity());
		if (entreeSortie == 0) {
			this.myLogger.config("Création d'un message de diffusion pour une entrée de voiture par la porte : " + this.myDoor.getIdentity() + " avec estampille : " + this.estampilleLocale);
		} else {
			this.myLogger.config("Création d'un message de diffusion pour une sortie de voiture par la porte : " + this.myDoor.getIdentity() + " avec estampille : " + this.estampilleLocale);
		}
		
		//increment of the internal clock
		this.estampilleLocale++;
		try {
			this.identifiantsPortes = myReso.getClients();
			this.myLogger.info("[PORTE" + this.myDoor.getIdentity() + "] : Envoi diffusion | Type : " + entreeSortie + " | Estampille locale : " + (estampilleLocale - 1));
			//message is diffused to all the processes within the group
			for (Integer porte : identifiantsPortes) {
				myReso.sendMessage(myDoor.getIdentity(), porte.intValue(), msgDiffusion);
			}
			
			//creation of a manager to manage the reception of acknowledgments
			this.gestionnaireRetourAck.add(new RetourAck(msgDiffusion.getIdMessage(), this.identifiantsPortes.size()));
		} catch (RemoteException e) {
			e.printStackTrace();
		}
	}
	
	/**
	 * Method executed on reception of a message by a door
	 * @param md : Object containing the clock.ID value, the type of request, identity of the original
	 * message and sender's identity
	 */
	public synchronized void recevoirMessageDiffusion(MessageDiffusion md) {
		this.myLogger.info("[PORTE" + this.myDoor.getIdentity() + "] <- Message diffusion | ID : " + md.getIdMessage());
		
		//increment of the local clock
		this.estampilleLocale++;
		String estID = this.estampilleLocale + "." + myDoor.getIdentity();
		//new calculated value estampille.idEmetteur
		Float newEstIdEm = new Float(estID);
		//assignment of the new calculated clock.id to the received message
		md.setEstIdEm(newEstIdEm.floatValue());
		
		synchronized (listeMessages) {
			listeMessages.add(md);
		}
		
		//sending of the acknowledgment message
		this.envoieAcknowledgementEmetteur(md);
	}
	
	public synchronized void envoieAcknowledgementEmetteur(MessageDiffusion md) {
		int emetteurOrigine = md.getEmetteurOrigine();
		int positionMD = cherchePositionMessageDiffusion(md.getIdMessage());
		float ackf = this.listeMessages.get(positionMD).getEstIdEm();
		
		//creation of the acknowledgment message with the new clock.id
		MessageAcknowledgment ma = new MessageAcknowledgment(md.getIdMessage(), ackf);
		
		try {
			//sending the acknowledgment message
			myReso.sendMessage(myDoor.getIdentity(), emetteurOrigine, ma);
		} catch (RemoteException e) {
			e.printStackTrace();
		}
		
		this.myLogger.info("[PORTE" + this.myDoor.getIdentity() + "] -> " + emetteurOrigine + " | ACK");
	}
	
	/**
	 * Method executed on reception of an acknowledgment message
	 * @param ma : The acknowledgment message containing the value clock.id and the ID of the original message
	 */
	public synchronized void recevoirAcknowledgement(MessageAcknowledgment ma) {
		this.myLogger.info("[PORTE" + this.myDoor.getIdentity() + "] <- Réception ACK | Message origine : " + ma.getIdMessage() + " | Val ACK : " + ma.getEstampilleProcessus());
		
		//in search of the original message in the manager of acknowledgment messages
		for (RetourAck ra : gestionnaireRetourAck) {
			if (ra.getIdMessage().compareTo(ma.getIdMessage()) == 0) {
				//original message has been found
				//adding of the received timestamp in the table reserved for the original message
				ra.getRetourAck().add(ma.getEstampilleProcessus());
				
				//if the size of the table reserved for the original message is equal to the number of clients
				//in the group, it means that all the acknowledgment messages have been received 
				if (ra.getRetourAck().size() == ra.getNbClients()) {
					
					//in search of the maximum value in the table
					Float max = Collections.max(ra.getRetourAck());
					
					this.myLogger.info("[PORTE" + this.myDoor.getIdentity() + "] : Diffusion Finale | Val finale : " + max + " | ID Msg : " + ma.getIdMessage());
					//creation of the final message
					MessageFinalDiffusion mf = new MessageFinalDiffusion(ma.getIdMessage(), max.floatValue());
					for (Integer porte : identifiantsPortes) {
						try {
							//diffusion of the final message
							myReso.sendMessage(myDoor.getIdentity(), porte.intValue(), mf);
						} catch (RemoteException e) {
							e.printStackTrace();
						}
					}
					//all acknowledgments have been received for this message
					//if can now be removed from the manager
					this.gestionnaireRetourAck.remove(ra);
				}
				//orginal message has been found and treated, no need to explore the rest of the list
				break;
			}
		}
	}
	
	/**
	 * Method used to search for the position of a message in the list of all received messages
	 * @param idMessage : ID of the message in search of the position in the list of all received messages
	 * @return : Position of the message in the list
	 */
	private int cherchePositionMessageDiffusion(String idMessage) {
		int i = 0;		
		
		while (i < this.listeMessages.size()) {
			if (this.listeMessages.get(i).getIdMessage().compareTo(idMessage) == 0) {
				//message found
				return i;
			}
			i++;
		}

		//message hasn't been found
		return -1;
	}
	
	/**
	 * Method executed on reception of a final diffusion for a message previously received and waiting
	 * to be delivered. If a message is found on top of the list of all received messages and it has READY as
	 * status, it is then delivered
	 * @param msg : Object containing the ID of the message and the final value clock.id for this message
	 */
	public synchronized void traiteDiffusionFinale(MessageFinalDiffusion msg) {
		this.myLogger.info("[PORTE" + this.myDoor.getIdentity() + "] <- Diffusion Finale | Msg : " + msg.getIdMessage());
		//in search of the position of the original message in the list of all received messages
		int positionMD = cherchePositionMessageDiffusion(msg.getIdMessage());
		
		if (positionMD != -1) {
			//position found
			//update of the original message with the new clock.id and setting the status to READY
			synchronized (listeMessages) {
				this.listeMessages.get(positionMD).setEstIdEm(msg.getDiffusionMax());
				this.listeMessages.get(positionMD).setMessageStatus(Message.READY);
				
				//sorting of the list
				Collections.sort(this.listeMessages);
			}
			
		}
		
		MessageDiffusion mdAExecuter = null;
		
		while (this.listeMessages.get(0).getMessageStatus() == Message.READY) {
			//a message is ready to be delivered
			synchronized (listeMessages) {
				mdAExecuter = this.listeMessages.get(0);
			}
			
			if ((mdAExecuter.getTypeMessage() == 0) && (mdAExecuter != null)) {
				//the message to be delivered is an entry of a car in the parking
				synchronized (carManager) {
					if (this.carManager.getNbCars() > 0) {
						if (mdAExecuter.getEmetteurOrigine() == this.myDoor.getIdentity()) {
							//door is the initiator, car has entered the parking
							this.carManager.decNbCars();
							this.myLogger.info("[PORTE" + this.myDoor.getIdentity() + "] : Voiture entrée | Avant : " + (this.carManager.getNbCars() + 1) + " | Après : " + this.carManager.getNbCars());
						} else {
							//door isn't the initiator, but there is at least one spot available in the parking
							//car has entered via another door
							this.carManager.decNbCars();
							this.myLogger.info("[PORTE" + this.myDoor.getIdentity() + "] : Voiture entrée | Autre porte : " + mdAExecuter.getEmetteurOrigine() + " | Avant : " + (this.carManager.getNbCars() + 1) + " | Après : " + this.carManager.getNbCars());
						}
					} else {
						//no more spot available, car has been refused
						this.myLogger.info("[PORTE" + this.myDoor.getIdentity() + "] : Refus | Places : " + this.carManager.getNbCars() + " | Msg : " + mdAExecuter.getIdMessage());
					}
				}
			} else if (mdAExecuter.getTypeMessage() == 1) {
				//request is an exit of a vehicle
				synchronized (carManager) {
					if (mdAExecuter.getEmetteurOrigine() == this.myDoor.getIdentity()) {
						//door is the initiator, increase of the number of spots available
						this.carManager.incNbCars();
						this.myLogger.info("[PORTE" + this.myDoor.getIdentity() + "] : Voiture sortie | Avant : " + (this.carManager.getNbCars() - 1) + " | Après : " + this.carManager.getNbCars());
					} else {
						//door is not the initiator, car has exited through another door, increase of the number of available spots
						this.carManager.incNbCars();
						this.myLogger.info("[PORTE" + this.myDoor.getIdentity() + "] : Voiture sortie | Autre porte : " + mdAExecuter.getEmetteurOrigine() + " | Avant : " + (this.carManager.getNbCars() - 1) + " | Après : " + this.carManager.getNbCars());
					}
				}
			}
			
			synchronized (listeMessages) {
				//message has been treated and can thus be deleted from the list
				this.listeMessages.remove(mdAExecuter);
			}
			
		}
	}

	/**
	 * Overridden method of the Client interface
	 * Messages received is either of type MessageAcknowledgment, MessageDiffusion or MessageFinalDiffusion
	 */
	@Override
	public void receiveMessage(int from, int to, Serializable msg) throws RemoteException {
		if (msg instanceof MessageAcknowledgment) {
			//an acknowledgment has been received
			this.recevoirAcknowledgement((MessageAcknowledgment) msg);
		} else if (msg instanceof MessageDiffusion) {
			//a message hass been received, it will be stored but not delivered immediately
			this.recevoirMessageDiffusion((MessageDiffusion) msg);
		} else if (msg instanceof MessageFinalDiffusion) {
			//receipt of the final value clock.id
			this.traiteDiffusionFinale((MessageFinalDiffusion) msg);
		}
	}

	/**
	 * Overridden method of the IAlgorithm interface
	 * Used for the request of car wishing to enter the parking
	 * Notice that a thread is started for a car wishing to enter the parking. This is done in order to be
	 * more efficient. Indeed, if on one door i have two consecutive entries. What will happen is that the first
	 * car will execute the envoieDiffusionGroupe method. The second will be blocked as it is a synchronized 
	 * method. When the first car has finished executing this method, the second will execute it. Neither of
	 * the cars will enter the parking if the final diffusion hasn't been calculated yet. The interesting fact
	 * about using threads in this algorithm is that meanwhile the first car has started receiving
	 * acknowledgments, the second can start the diffusion process. One important point to pin point out is that
	 * they share the list of all received messages. So whenever using this list, the use of the latter has to
	 * be thread safe. This problem is solved as all methods are synchronized.
	 */
	@Override
	public void demandeEntree() {
		Thread demandeEntree = new Thread(this, "ThreadEntree");
		demandeEntree.start();
	}

	/**
	 * Overridden method of the IAlgorithm interface
	 * Used for the request of car wishing to exit the parking
	 * The explanation on the use of threads is the same as in the demandeEntree method but this time on an exit
	 * of a vehicle
	 */
	@Override
	public void demandeSortie() {
		Thread demandeSortie = new Thread(this, "ThreadSortie");
		demandeSortie.start();
	}

	/**
	 * Overridden method of the Runnable interface
	 * A thread is started according to an entry or an exit of a vehicle in the parking
	 */
	@Override
	public void run() {
		String thName = Thread.currentThread().getName();
		
		if (thName.equals("ThreadEntree")) {
			this.myLogger.config("-!- Création et lancement d'un Thread pour gérer l'entrée de la voiture...");
			this.envoieDiffusionGroupe(0);
		} else if (thName.equals("ThreadSortie")) {
			this.myLogger.config("-!- Création et lancement d'un Thread pour gérer la sortie de la voiture...");
			this.envoieDiffusionGroupe(1);
		}
	}

}
